package com.xinqing.test.flink.processfunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;

/**
 * @Author:CHQ
 * @Date:2021/4/15 10:43
 * @Description
 */
public class FraudDetectionJob {
    public static void main(String args[]) {
        try {
            //env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            //source
            DataStreamSource<Transaction> source = env.addSource(new TransactionSource(), "transactions");

            //transform
            SingleOutputStreamOperator<Alert> process = source.keyBy(Transaction::getAccountId).process(new FraudDetector()).name("process");

            //sink
            process.addSink(new AlertSink()).name("send alert");

            //execute
            env.execute("Fraud Detection");
        } catch (Exception e) {
            System.out.println(e);
        }

    }
}
